use crate::config::app_error::{AppError, AppResult};
use crate::extension::index::index::query::index_descriptor::IndexDescriptor;
use crate::extension::index::index::query::index_entry::IndexEntry;
use crate::extension::index::index::query::index_spec::IndexSpec;
use std::any::{Any, TypeId};
use std::cell::RefCell;
use std::collections::HashMap;
use std::hash::Hash;
use std::sync::Arc;
use std::sync::RwLock;

pub struct IndexEntryContainer {
    index_entries: RwLock<HashMap<String, (TypeId, Box<dyn Any>)>>,
}

impl IndexEntryContainer {
    pub fn new() -> Self {
        IndexEntryContainer {
            index_entries: RwLock::new(HashMap::new()),
        }
    }

    /**
     * Add an `IndexEntry` to this container.
     *
     * @param entry the entry to add
     * @throws if the entry already exists
     */
    pub fn add<IE: IndexEntry + 'static>(&self, index_entry: Arc<RefCell<IE>>) -> AppResult<()> {
        let type_id = TypeId::of::<IE>();

        let index_descriptor = index_entry.borrow().get_index_descriptor().identity();

        let mut index_entries = self.index_entries.write().map_err(|_| {
            AppError::InternalServerError(
                "Failed to acquire write lock on index_entries".to_string(),
            )
        })?;

        if index_entries.contains_key(&index_descriptor) {
            return Err(
                anyhow::anyhow!("Index entry already exists for {:?}", index_descriptor).into(),
            );
        }
        index_entries.insert(index_descriptor, (type_id, Box::new(index_entry)));
        Ok(())
    }

    /**
     * Get the `IndexEntry` for the given `IndexDescriptor`.
     *
     * @param index_descriptor the index descriptor
     * @return the index entry
     */
    pub fn get<IE: IndexEntry + 'static>(
        &self,
        index_descriptor: &IndexDescriptor<IE::T>,
    ) -> AppResult<Option<Arc<RefCell<IE>>>>
    where
        IE::T: Send + Sync + Hash + PartialEq,
    {
        let index_entries = self.index_entries.read().map_err(|_| {
            AppError::InternalServerError(
                "Failed to acquire read lock on index_entries".to_string(),
            )
        })?;

        if let Some((type_id, index_entry)) = index_entries.get(&index_descriptor.identity()) {
            if let Some(entry) = index_entry.downcast_ref::<Arc<RefCell<IE>>>() {
                Ok(Some(Arc::clone(entry)))
            } else {
                Ok(None)
            }
        } else {
            Ok(None)
        }
    }

    pub fn contains(&self, identity: &String) -> AppResult<bool> {
        let index_entries = self.index_entries.read().map_err(|_| {
            AppError::InternalServerError(
                "Failed to acquire read lock on index_entries".to_string(),
            )
        })?;

        Ok(index_entries.contains_key(identity))
    }

    pub fn remove<IE: IndexEntry+ 'static>(
        &self,
        identity: &String,
    ) -> AppResult<Option<Arc<RefCell<IE>>>>
    where
        IE::T: Send + Sync + Hash + PartialEq,
    {
        let mut index_entries = self.index_entries.write().map_err(|_| {
            AppError::InternalServerError(
                "Failed to acquire write lock on index_entries".to_string(),
            )
        })?;

        if let Some((type_id, index_entry)) = index_entries.remove(identity) {
            if let Some(entry_ref) = index_entry.downcast_ref::<Arc<RefCell<IE>>>() {
                Ok(Some(Arc::clone(entry_ref)))
            } else {
                Ok(None)
            }
        } else {
            Ok(None)
        }
    }

    pub fn size(&self) -> AppResult<usize> {
        let mut index_entries = self.index_entries.read().map_err(|_| {
            AppError::InternalServerError(
                "Failed to acquire read lock on index_entries".to_string(),
            )
        })?;
        Ok(index_entries.len())
    }

    pub fn for_each<F, IE>(&self, f: F) -> AppResult<()>
    where
        IE: IndexEntry + 'static,
        IE::T: Send + Sync + Hash + PartialEq,
        F: Fn(&Arc<RefCell<IE>>),
    {
        let mut index_entries = self.index_entries.read().map_err(|_| {
            AppError::InternalServerError(
                "Failed to acquire read lock on index_entries".to_string(),
            )
        })?;
        Ok(index_entries
            .iter()
            .for_each(|(_, (type_id, index_entry))| {
                if let Some(entry_ref) = index_entry.downcast_ref::<Arc<RefCell<IE>>>() {
                    f(entry_ref);
                }
            }))
    }

    pub fn iter<IE: IndexEntry + 'static>(&self) -> AppResult<Vec<Arc<RefCell<IE>>>> {
        let mut index_entries = self.index_entries.read().map_err(|_| {
            AppError::InternalServerError(
                "Failed to acquire read lock on index_entries".to_string(),
            )
        })?;
        let mut result = Vec::new();
        for (name, (type_id, index_entry)) in index_entries.iter() {
            if let Some(entry) = index_entry.downcast_ref::<Arc<RefCell<IE>>>() {
                result.push(Arc::clone(entry));
            }
        }
        Ok(result)
    }

    pub fn remove_fn<F, IE>(&mut self, f: F) -> AppResult<Vec<Arc<RefCell<IE>>>>
    where
        IE: IndexEntry + 'static,
        F: Fn(&IndexDescriptor<IE::T>) -> bool,
    {
        let mut index_entries = self.index_entries.write().map_err(|_| {
            AppError::InternalServerError(
                "Failed to acquire write lock on index_entries".to_string(),
            )
        })?;

        let to_remove = index_entries
            .iter()
            .filter_map(|(key, (type_id, index_entry))| {
                if let Some(entry_ref) = index_entry.downcast_ref::<Arc<RefCell<IE>>>() {
                    if f(&entry_ref.borrow().get_index_descriptor()) {
                        Some(key.to_owned())
                    } else {
                        None
                    }
                } else {
                    None
                }
            })
            .collect::<Vec<String>>();

        let mut vec = Vec::new();
        for key in to_remove {
            if let Some((type_id, index_entry)) = index_entries.remove(&key) {
                if let Some(entry_ref) = index_entry.downcast_ref::<Arc<RefCell<IE>>>() {
                    vec.push(Arc::clone(entry_ref));
                }
            }
        }
        Ok(vec)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        extension::index::index::query::index_entry::IndexEntryImpl, tests::primary_key_index_spec,
    };

    #[test]
    fn add() {
        let mut container = IndexEntryContainer::new();

        let spec = primary_key_index_spec();
        let descriptor = IndexDescriptor::new(spec);
        let entry = IndexEntryImpl::new(descriptor);

        let entry = Arc::new(RefCell::new(entry));
        container.add(entry.clone()).unwrap();

        let spec = primary_key_index_spec();
        let descriptor = IndexDescriptor::new(spec);
        assert!(container.contains(&descriptor.identity()).unwrap());

        let result = container.add(entry.clone());
        assert!(result.is_err());
        assert_eq!(
            result.unwrap_err().to_string(),
            format!(
                "error:`Index entry already exists for {:?}`",
                descriptor.identity()
            )
        );
    }

    #[test]
    fn remove() {
        let mut container = IndexEntryContainer::new();

        let spec = primary_key_index_spec();
        let descriptor = IndexDescriptor::new(spec);
        let entry = IndexEntryImpl::new(descriptor);

        let entry = Arc::new(RefCell::new(entry));
        container.add(entry.clone()).unwrap();

        let spec = primary_key_index_spec();
        let descriptor = IndexDescriptor::new(spec);
        assert!(container.contains(&descriptor.identity()).unwrap());
        assert_eq!(container.size().unwrap(), 1);

        container.remove::<IndexEntryImpl<FakeExtension>>(&descriptor.identity()).unwrap();

        assert!(!container.contains(&descriptor.identity()).unwrap());
        assert_eq!(container.size().unwrap(), 0);
    }
}
